Cloud DLP で検出された機密データが格納されている場所を BigQuery で特定してみた

エノカワ

2022.03.25

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは！エノカワです。

Google Cloud の DLP（Data Loss Prevention）サービスである Cloud DLP では、
GCS や BigQuery、Datastore などに格納済みのデータに機密データが含まれるかを検査することできます。

下記エントリで Cloud DLP を使った BigQuery の個人情報の検出が紹介されているので是非ご参照ください。

では、Cloud DLP で検査して BigQuery に機密データが含まれることが分かりました。となった場合に　　「どのテーブルのどの行に機密データが含まれているのか？」を確認することができるのでしょうか？

今回は Cloud DLP で検出された機密データが格納されている場所を BigQuery で特定することにチャレンジしてみました。

検査対象テーブル

BigQuery に Cloud DLP の検査対象となるテーブルを用意します。

氏名、県名、メールアドレスを含むフィールドを持つテーブルtextsを作成してみました。
通常のテキストの中に機密データを混ぜ込みたかったので、Python の Faker で作成したダミーデータをアレンジしています。

Welcome to Faker’s documentation!

検査ジョブの作成

検査対象テーブルから [エクスポート] > [DLP でスキャン] をクリックします。
Cloud DLP ジョブの作成ページが表示されるので、BigQueryテーブルの検査ジョブを設定しています。

入力データを選択

検査対象データの情報を入力していきます。

「名前」には任意のジョブ名を入力します。
「リソースロケーション」にはデフォルトで「グローバル（任意のリージョン）」が選択されているのでそのままとします。
「場所」には検査対象テーブルtextsの情報を入力します。

「サンプリング」では入力データの一部のみ検査するように設定できますが、
今回は検査対象データが少ないので「サンプリングなし」とします。

フィールドに関する下記の設定を入力します。

フィールドの識別（カンマ区切り）
id
検査する列
inspect only specified columns
フィールド（カンマ区切り）
text

2. 3. で機密データを含むtextフィールドのみを検査対象として指定しています。

1. の識別フィールドにはテーブル内で一意の行識別子idフィールドを指定しています。
今回は検査対象テーブルと検査結果テーブルを結合する際の結合キーとしても使用します。
なお、識別フィールドに値が含まれている行のみが検査対象となるようです。

検出の設定

検出対象の機密データの種類（INFOTYPE）を設定しています。

「INFOTYPEを設定」をクリックして、下記の3種類を設定します。

PRESON_NAME：人の名前
EMAIL_ADDRESS：メールアドレス
LOCATION：物理的な住所または場所

その他の設定はデフォルトのままとします。

アクションの追加

ジョブの完了後に実行させるアクションを設定していきます。

複数のアクションを設定することができますが、今回は「BigQueryに保存」のみを有効にします。

検査結果を保存するテーブルtexts_dlpの情報を入力します。
検査ジョブ完了時に自動でテーブルが作成されるので事前に作成しておく必要はありません。

「見積もりを含める」を有効にします。
有効にすると検出された機密データも BigQuery に保存されます。
今回は検証のために検出された機密データも確認したいので有効にします。

スケジュール

スケジュールオプションを選択します。

検査を定期的に実行する場合にスケジュールを設定できますが、
今回は検証のために1回限りの実行としたいので「なし」を選択します。

確認

最後にジョブ構成を確認します。

指定したジョブ設定の概要が JSON 形式で表示されます。
赤枠で囲った箇所が「入力データを選択」セクションで行ったフィールドに関する設定です。

identifyingFields：識別フィールド
includedFields：検査フィールド

この後に行う検査対象テーブルと検査結果を結合でポイントになってくる設定です。

「作成」ボタンをクリックすると、確認ダイアログが表示されます。
検査するデータの量に応じて費用が急増する可能性がある旨の確認ですが、
今回は検査対象データが少ないので「作成を確認」をクリックします。

検査ジョブの結果

ジョブの詳細画面に遷移後、しばらくすると検査ジョブが完了しました！
14件検出されたようです。

画面下部の「ジョブの結果」に検査結果が表示されています。
検出された機密データの種類と件数は確認できますが、具体的にどういった機密データが検出されたのでしょうか？

詳細を確認するために「検出をBIGQUERYで表示」ボタンをクリックします。

検査結果が保存されたテーブルtexts_dlpの画面に遷移しました。

検査結果テーブルをクエリ

検査結果テーブルをクエリして詳細を確認してみましょう。

下記SQLでクエリを実行します。
REPEATEDなフィールドも含んでいるのでUNNESTで中身を展開しています。

SELECT
    quote,
    info_type.name,
    likelihood,
    CAST(id_values AS INT64) AS id,
    location.byte_range.start AS byte_range_start,
    location.byte_range.end AS byte_range_end
FROM
    dlp_work.texts_dlp
JOIN
    UNNEST(location.content_locations) AS _content_locations
JOIN
    UNNEST(_content_locations.record_location.record_key.id_values) AS id_values
;

検出された14件の内容や機密データの種類、場所が分かりました！

quoteは実際に検出された内容です。
ちなみに、検査ジョブ作成時のアクションの追加セクションで指定した「見積もりを含める」を無効にすると値が一律nullとなります。（機密データが BigQuery に保存されなくなる仕組み）

name（info_type.name）は機密データの種類、likelihoodは一致の可能性を示しています。

byte_range_start（location.byte_range.start）とbyte_range_end（location.byte_range.end）は
それぞれ検出された場所の開始位置と終了位置を示しており、ゼロベースのバイトオフセットの値です。　　

id（id_values）は識別フィールドに指定したidの値です。

それでは、検査対象テーブルtextsと検査結果テーブルtexts_dlpを結合してみましょう。
識別フィールドidを結合キーにして内部結合します。

WITH texts_dlp AS (
    SELECT
        quote,
        info_type.name,
        likelihood,
        CAST(id_values AS INT64) AS id,
        location.byte_range.start AS byte_range_start,
        location.byte_range.end AS byte_range_end
    FROM
        dlp_work.texts_dlp
    JOIN
        UNNEST(location.content_locations) AS _content_locations
    JOIN
        UNNEST(_content_locations.record_location.record_key.id_values) AS id_values
)

SELECT
    texts.*,
    dlp.* EXCEPT(id)
FROM
    dlp_work.texts AS texts
JOIN
    texts_dlp AS dlp
USING (id)
;

検出された内容を含む行のみ抽出されました！

同一のtextから複数のquoteが検出されていることが分かりますね。
機密データの種類や一致の可能性、検出された場所の開始位置と終了位置も合わせて確認することができます。

検査結果テーブルを外部結合することで各行で検出された件数を集計することもできました。

WITH texts_dlp AS (
    SELECT
        quote,
        info_type.name,
        likelihood,
        CAST(id_values AS INT64) AS id,
        location.byte_range.start,
        location.byte_range.end
    FROM
        dlp_work.texts_dlp
    JOIN
        UNNEST(location.content_locations) AS _content_locations
    JOIN
        UNNEST(_content_locations.record_location.record_key.id_values) AS id_values
)

SELECT
    texts.id,
    texts.text,
    COUNT(dlp.id) AS count
FROM
    dlp_work.texts AS texts
LEFT JOIN
    texts_dlp AS dlp
USING (id)
GROUP BY 1, 2
;